{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "0",
   "metadata": {},
   "source": [
    "# Handling errors with the RunResponseProtocol"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "from dotenv import load_dotenv\n",
    "\n",
    "from autogen import ConversableAgent, LLMConfig\n",
    "from autogen.events.agent_events import ExecutedFunctionEvent\n",
    "from autogen.io.processors.base import EventProcessorProtocol\n",
    "from autogen.io.processors.console_event_processor import ConsoleEventProcessor\n",
    "from autogen.io.run_response import RunResponseProtocol\n",
    "from autogen.tools import tool\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "2",
   "metadata": {},
   "source": [
    "## Handling custom LLM errors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# use this instead of the default response.process() to handle custom error messages\n",
    "def custom_response_process(\n",
    "    response: RunResponseProtocol,\n",
    "    processor: EventProcessorProtocol | None = None,\n",
    "    fail_on_error_messages_list: list[str] | None = None,\n",
    ") -> None:\n",
    "    processor = processor or ConsoleEventProcessor()\n",
    "    for event in response.events:\n",
    "        if (\n",
    "            fail_on_error_messages_list\n",
    "            and isinstance(event, ExecutedFunctionEvent)\n",
    "            and not event.content.is_exec_success\n",
    "            and any(error_msg in event.content.content for error_msg in fail_on_error_messages_list)\n",
    "        ):\n",
    "            raise RuntimeError(f\"Function execution failed: {event.content}\")\n",
    "        processor.process_event(event)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4",
   "metadata": {},
   "outputs": [],
   "source": [
    "llm_config = LLMConfig(\n",
    "    config_list={\"api_type\": \"openai\", \"model\": \"gpt-5-nano\", \"api_key\": os.getenv(\"OPENAI_API_KEY\")}\n",
    ")\n",
    "\n",
    "for error_msg in [\"Some other error message\", \"This function is not implemented yet.\"]:\n",
    "    agent = ConversableAgent(\n",
    "        name=\"agent\",\n",
    "        llm_config=llm_config,\n",
    "    )\n",
    "\n",
    "    @tool(description=\"List files and folders\")\n",
    "    def list_files(\n",
    "        folder_name: str,\n",
    "    ) -> str:\n",
    "        raise NotImplementedError(\"This function is not implemented yet.\")\n",
    "\n",
    "    list_files.register_for_llm(agent)\n",
    "\n",
    "    response = agent.run(\n",
    "        message=\"List all files and folders in the 'root' folder\",\n",
    "        tools=agent.tools,\n",
    "        user_input=False,\n",
    "        max_turns=3,\n",
    "    )\n",
    "\n",
    "    # The tool will raise NotImplementedError(\"This function is not implemented yet.\") and we won't kill the process\n",
    "    custom_response_process(\n",
    "        response=response,\n",
    "        fail_on_error_messages_list=[error_msg],\n",
    "    )\n",
    "    print(f\"Summary: {response.summary}\")\n",
    "    print(\"*\" * 40 + \"\\nSUCCESS\\n\" + \"*\" * 40)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "5",
   "metadata": {},
   "source": [
    "## Inner agents exceptions - team within a tool"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6",
   "metadata": {},
   "outputs": [],
   "source": [
    "llm_config = LLMConfig(\n",
    "    config_list={\"api_type\": \"openai\", \"model\": \"gpt-5-nano\", \"api_key\": os.getenv(\"OPENAI_API_KEY\")}\n",
    ")\n",
    "\n",
    "agent = ConversableAgent(\n",
    "    name=\"agent\",\n",
    "    llm_config=llm_config,\n",
    ")\n",
    "\n",
    "\n",
    "@tool(description=\"List files and folders\")\n",
    "def list_files(\n",
    "    folder_name: str,\n",
    ") -> str:\n",
    "    # llm_config = LLMConfig(config_list={\"api_type\": \"openai\", \"model\": \"gpt-5-nano\", \"api_key\": \"abc\"})\n",
    "    llm_config = LLMConfig(config_list={\"api_type\": \"google\", \"model\": \"gemini-2.0-flash\", \"api_key\": \"abc\"})\n",
    "\n",
    "    agent = ConversableAgent(\n",
    "        name=\"agent\",\n",
    "        llm_config=llm_config,\n",
    "    )\n",
    "    response = agent.run(\n",
    "        message=\"List all files and folders in the 'root' folder\",\n",
    "        tools=agent.tools,\n",
    "        user_input=False,\n",
    "        max_turns=3,\n",
    "    )\n",
    "    response.process()\n",
    "    return response.summary\n",
    "\n",
    "\n",
    "list_files.register_for_llm(agent)\n",
    "\n",
    "response = agent.run(\n",
    "    message=\"List all files and folders in the 'root' folder\",\n",
    "    tools=agent.tools,\n",
    "    user_input=False,\n",
    "    max_turns=3,\n",
    ")\n",
    "\n",
    "fail_on_error_messages_list = [\n",
    "    \"Incorrect API key provided\",  # openai\n",
    "    \"API key not valid. Please pass a valid API key.\",  # gemini\n",
    "]\n",
    "\n",
    "# event will contain \"content=\"Error: Error code: 401 - {'error': {'message': 'Incorrect API key provided: abc. You can find your API key ....\"\n",
    "# and error will be raised\n",
    "custom_response_process(response=response, fail_on_error_messages_list=fail_on_error_messages_list)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "7",
   "metadata": {},
   "source": [
    "## Async version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# use this instead of the default await response.process() to handle custom error messages\n",
    "from autogen.io.processors.console_event_processor import AsyncConsoleEventProcessor\n",
    "from autogen.io.run_response import AsyncRunResponseProtocol\n",
    "\n",
    "\n",
    "async def a_custom_response_process(\n",
    "    response: AsyncRunResponseProtocol,\n",
    "    processor: AsyncConsoleEventProcessor | None = None,\n",
    "    fail_on_error_messages_list: list[str] | None = None,\n",
    ") -> None:\n",
    "    processor = processor or AsyncConsoleEventProcessor()\n",
    "    async for event in response.events:\n",
    "        if (\n",
    "            fail_on_error_messages_list\n",
    "            and isinstance(event, ExecutedFunctionEvent)\n",
    "            and not event.content.is_exec_success\n",
    "            and any(error_msg in event.content.content for error_msg in fail_on_error_messages_list)\n",
    "        ):\n",
    "            raise RuntimeError(f\"Function execution failed: {event.content}\")\n",
    "        await processor.process_event(event)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9",
   "metadata": {},
   "outputs": [],
   "source": [
    "llm_config = LLMConfig(\n",
    "    config_list={\"api_type\": \"openai\", \"model\": \"gpt-5-nano\", \"api_key\": os.getenv(\"OPENAI_API_KEY\")}\n",
    ")\n",
    "\n",
    "for error_msg in [\"Some other error message\", \"This function is not implemented yet.\"]:\n",
    "    agent = ConversableAgent(\n",
    "        name=\"agent\",\n",
    "        llm_config=llm_config,\n",
    "    )\n",
    "\n",
    "    @tool(description=\"List files and folders\")\n",
    "    def list_files(\n",
    "        folder_name: str,\n",
    "    ) -> str:\n",
    "        raise NotImplementedError(\"This function is not implemented yet.\")\n",
    "\n",
    "    list_files.register_for_llm(agent)\n",
    "\n",
    "    response = await agent.a_run(\n",
    "        message=\"List all files and folders in the 'root' folder\",\n",
    "        tools=agent.tools,\n",
    "        user_input=False,\n",
    "        max_turns=3,\n",
    "    )\n",
    "\n",
    "    await a_custom_response_process(\n",
    "        response=response,\n",
    "        fail_on_error_messages_list=[error_msg],\n",
    "    )\n",
    "    print(f\"Summary: {await response.summary}\")\n",
    "    print(\"*\" * 40 + \"\\nSUCCESS\\n\" + \"*\" * 40)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "10",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "front_matter": {
   "description": "This notebook demonstrates handling errors with the RunResponseProtocol",
   "tags": [
    "errors",
    "runresponseprotocol"
   ]
  },
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
